package com.cmbyc.FlinkSQL.Test


import org.apache.flink.streaming.api.scala._
import org.apache.flink.table.api.{DataTypes, Table}
import org.apache.flink.table.api.bridge.scala.{StreamTableEnvironment, tableConversions}
import org.apache.flink.table.descriptors.{Csv,Kafka, Schema}



/**
 *
 * @program: com.cmbyc.FlinkSQL.Test
 * @author: YCLW058
 * @create: 2021-05-20 10:10
 * @decsription:
 *
 * */

object Demo02 {
  def main(args: Array[String]): Unit = {
    //1 env
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)
    val tenv = StreamTableEnvironment.create(env)

    //2 source
    //2.1
    //文件路径
//    val filepath="D:\\tool\\server\\IDEA\\code\\Java\\Flink\\Flink_scala\\data\\input\\test"
//    //注册表 定义表结构
//    tenv.connect(new FileSystem().path(filepath))
//      .withFormat(new OldCsv)
//      .withSchema(new Schema()
//      .field("id",DataTypes.INT())
//        .field("name",DataTypes.STRING())
//        .field("age",DataTypes.INT())
//      ).createTemporaryTable("inputTable")

    // 2.2 从kafka读取数据
    tenv.connect(new Kafka()
    .version("2.8.0")
      .topic("hello")
      .property("zookeeper.connect","localhost:2181")
      .property("bootstrp.sercers","localhost:9092")
    )
      .withFormat(new Csv())
      .withSchema(new Schema()
        .field("id",DataTypes.INT())
        .field("name",DataTypes.STRING())
        .field("age",DataTypes.INT())
      )
      .createTemporaryTable("kafkainputTable")

    //3 transformation

    //4 sink
    val StudentTable: Table =tenv.from("kafkainputTable")
    StudentTable.toAppendStream[(Int,String,Int)].print()
    //5 execute
    env.execute()

  }

}
